
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：data_cache缓存服务器连接到redis(GaussDb)时使用(data_cache专用)
--- data_cache 使用ngx自带的长连接，连接到redis(GaussDb)数据库
--- dc读写redis的封装，其他业务应该不会使用到这个文件
---
--- 业务服务器 ----> data_cache缓存服务器 --(redisMgrDc.lua)--> redis(GaussDb)数据库
----------------------------------------------------------------


local assert     = assert
local require    = require
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

require "resty.core.shdict"

local ngx_shared  = ngx.shared;
local string_sub  = string.sub
local string_gsub = string.gsub
local string_format = string.format

local log       = require("modules.commhm.log");
local ns_time   = require("modules.commhm.time");
local uu        = require("modules.commhm.utils");

local ns_redis  = require("resty.redis");

local ns_semaphore = require("ngx.semaphore")


local table_insert = table.insert
local table_remove = table.remove
local ngx_sleep = ngx.sleep
local ngx_now   = ngx.now

local NGX_NULL  = "not_found"                  --ngx.null
local NGX_AUTH  = "authentication required."


-------------------------------------------------------------------------------


local _REDIS_DICT_ADD_LOCK_  = false     --shared_dict add检查锁(乐观锁)
local _REDIS_RESTY_LOCK      = true      --shared lock


local __TJ_INTERVAL__      = 180


local __DEFAULT_READ_LIMIT__ = 256      ---默认最大读队列
local __DEFAULT_SAVE_LIMIT__ = 312      ---默认最大写队列 优先保护写
local __BUSY_BLOCK_LIMIT__   = 64       ---限流次数


local __SUCC__ = 0    ---返回值 成功
local __FAIL__ = 1    ---返回值 失败


-- REDIS MGR类
-- _VERSION = '0.12.1'
local _M = {}
local mt = { __index = _M }



function _M.new(self, pre_, config_server_, config_select_ )
	log.debug( log.fast() and "call redisMgr:new" );
	local ins_ = {
		config_cache  = {};
		pre           = pre_;   --ma posting map

		select_mod    = (config_select_.select_mod or 10);  ---10分表  100分表

		config_server = config_server_;
		config_select = config_select_;

		read_semas_tj = {};
		read_only_redis_block = 0;    ---忙的时候流控 无法mread

		last_time_report_tj        = uu.now() + __TJ_INTERVAL__;

		get_set_tj   = {};
		-- get_call
		-- get_cache
		-- get_cache_stale
		-- get_redis
		-- set_redis
		-- delay_redis

		mread_tj = {
			call  = 0,
			get   = 0,
			gets  = 0,
			set   = 0,
		};

		_lock_id   = (ngx.worker.id() + 1) * 1000;    --lock_id
	};
    return setmetatable(ins_, mt);
end



_M.init_pool = function(self)
	if  ngx.ctx.use_redis then
		-- created
	else
		ngx.ctx.use_redis = true;
		ngx.ctx.redis_dc_connections  = {};    --本次建立的连接池
	end
end



_M.final = function(self)
	if  ngx.ctx.use_redis then
		self:close();
		self:do_report_tj();
	end

end


---------------------------------------------------
---增加统计项
_M.set_get_set_tj = function(self, hash_, name_)
	if     ngx.ctx.m_params.act then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.act
	elseif ngx.ctx.m_params.cmd then
		hash_ = hash_ .. "_" .. ngx.ctx.m_params.cmd
	end

	if  not self.get_set_tj[hash_] then
		self.get_set_tj[hash_] = {
			get_call  = 0,
			set_redis = 0,
		}
	end

	if  not self.get_set_tj[hash_][ name_ ] then
		self.get_set_tj[hash_][name_] = 0
	end
	self.get_set_tj[hash_][ name_ ] = self.get_set_tj[hash_][ name_ ] + 1
end


---汇总打印统计项
_M.do_report_tj = function(self)
	local now_ = uu.now()
	if  now_ > self.last_time_report_tj then
		self.last_time_report_tj = now_ + __TJ_INTERVAL__;

		--开始打印统计
		for hash_, v1 in pairs(self.get_set_tj) do
			if  v1.get_call + v1.set_redis > 0 then
				log.day_list_short( "redis", "cache_hit_redis", ngx.worker.id(), hash_, uu.to_str(self.get_set_tj) )
			end
		end
		self.get_set_tj = {}

		if  self.mread_tj.call > 0 then
			log.day_list_short( "redis", "cache_hit_redis", ngx.worker.id(), "mread", self.pre,
					self.mread_tj.call, self.mread_tj.get, self.mread_tj.gets, self.mread_tj.set )
			self.mread_tj = { call=0, get=0, gets=0, set=0 }
		end
	end
end



------------------------------------------------API---------------------------
--设置kv
_M.API_redis_set_kv = function( self, redis_select, key, value, et )
	log.debug( log.fast() and "call API_redis_set_kv, redis_select=" .. redis_select .. ", key=" .. key );

	--优先写入share cache
	--self:set_shared_cache_redis( redis_select, key .. "_kv", value, et )

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select, __DEFAULT_SAVE_LIMIT__ )
	if  lock_id_  == 0 then
		self:error_exit( "set_kv_lock_err" );
		return __FAIL__, "set_kv_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ok, err;

	ok, err = instance:set(key, "" .. value);
	if  err then
		log.day_list_short( "redis", "error_set_kv_set", redis_select, key, err );
		self:error_exit( err );
		return __FAIL__, err
	end

	if  et and type(et)=="number" and et >= 0 then
		ok, err = instance:expire(key, et)
		if  err then
			log.day_list_short( "redis",  "error_set_kv_expire", redis_select, key, err );
			self:error_exit( err );
			return __FAIL__, err
		end
	end

	self:read_unlock2(redis_select)
	self:set_get_set_tj( "kv", "set_kv" )

	return __SUCC__
end


--获得kv
_M.API_redis_get_kv = function( self, redis_select, key )
	log.debug( log.fast() and "call API_redis_get_kv, redis_select=" .. redis_select .. ", key=" .. key );

	--[[
		local ret2, stale_ = self:get_shared_cache_redis( redis_select, key .. "_kv" )
		if  ret2 then
			if  stale_ then
				log.debug( log.fast() and "ignore stale_ ret2kv cache hit." )
			else
				log.debug( log.fast() and "ret2kv cache hit." )
				return __SUCC__, ret2   --cache命中
			end
		end
	--]]

	self:set_get_set_tj( "kv", "get_kv_call" )

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "get_kv_lock_err" );
		return __FAIL__, "get_kv_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret, err = instance:get(key);
	self:read_unlock2(redis_select)

	if  err then
		self:set_get_set_tj( "kv", "get_kv_error1" )
		log.day_list_short( "redis",  "error_get_kv_get", redis_select, key, err );
		self:error_exit( err );
		return __FAIL__, err
	end

	--local ttl_ = instance:ttl( key );
	--log.debug( "ttl=" .. uu.to_str( ttl_ ) )
	--if  err then
		--log.day_list_short( "redis", "error_get_kv_ttl", key, err );
		--self:error_exit( "get err:" .. (err or "nil") );
		--return;
	--end

	if  type(ret) == 'string' then
		--- ok
		--[[
		if  ngx.ctx.no_shared_cache == 'y' then
			-- 不保存 scan使用
			log.day_list_short( "redis", "no_shared_cache", redis_select, key );
		else
			if  #ret > 0 then
				self:set_shared_cache_redis( redis_select, key .. "_kv", ret )
			else
				log.day_list_simple( "redis", "error_len0", redis_select, key, ret );
			end
		end
		--]]
		self:set_get_set_tj( "kv", "get_kv_ok" )

	elseif ret == ngx.null then
		ret = NGX_NULL
		self:set_get_set_tj( "kv", "get_kv_null" )
	else
		self:set_get_set_tj( "kv", "get_kv_error2" )
		-- 返回的一定是一个string 或者 ngx.null 否则是出错
		log.day_list_short( "redis", "error_get_kv_type", key, "ret_not_string", type(ret) );
		self:error_exit( "error_get_kv_type" .. type(ret) );
		return __FAIL__, "err_ret_type"
	end

	return __SUCC__, ret   ---成功
end




-------------------------------------------------------
---写数据 key  hash  value
_M.API_redis_save = function( self, redis_select, key, hash, value, redis_delay_, read_only_ )
	log.debug( log.fast() and "call API_redis_save, redis_select=" .. redis_select
			.. ", key=" .. key .. ", hash=" .. hash );

	local key_hash_ = key .. "_h" .. hash

	--优先写入share cache
	local ret_ = self:set_shared_cache_redis( redis_select, key_hash_, value )
	if  ret_ and tonumber(redis_delay_) == 1 then
		--log.day_list_short( "redis", "redis_delay", redis_select, key, hash )
		self:unlock_resty_lock( redis_select, key_hash_, read_only_ )
		self:set_get_set_tj( hash, "save_delay" )
		return  __SUCC__  --缓写成功
	end

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select, __DEFAULT_SAVE_LIMIT__ )
	if  lock_id_  == 0 then
		self:error_exit( "save_lock_err" )
		return __FAIL__, "save_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ok, err = instance:hset(key, hash, value);   --- ok=0 if succ
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis",  "hset_fail", redis_select, key, hash, err );
		self:error_exit( "hset err:" .. (err or "nil") );
		return __FAIL__, err
	else
		--log.day_list_short( "redis",  "hset_succ", redis_select, key, hash, #value );
		self:unlock_resty_lock( redis_select, key_hash_, read_only_ )
	end

	self:set_get_set_tj( hash, "save_ok" )

	return __SUCC__   ---成功
end




--- 进行redis key_hash锁检查
--- 业务一致性锁，相当于redis的远程锁
--- 同一个key-hash被锁的情况，实际上是非常稀少的
--- 这里复杂的检测逻辑不会对系统性能有较大影响
--- 0=成功   其他=失败
_M.check_redis_resty_lock = function( self, redis_select, key_hash_ )
	local dict_ = ngx_shared[ self.pre .. redis_select ]
	if  dict_ then
		--2秒中内循环试图锁定一个 key_hash_lk ( exists则等待 )
		local timeout_ = 2000   ---毫秒
		local step_    = 1      ---步长毫秒
		local cc_      = 0

		---计算锁id
		self._lock_id = self._lock_id + 1
		if  self._lock_id >= 100000 then self._lock_id = 1000 end

		while timeout_ > 0 do
			cc_ = cc_ + 1
			local ok, err = dict_:add( key_hash_ .. "_lk" , self._lock_id, 3)      ---锁住主键kh  12345_6_lk
			if  ok then
				if  cc_ > 10 then
					--循环等待其他锁的次数太多(告警日志)
					log.day_list_short( "redis", "lock_rw", "lock_ok0", ngx.worker.id(), uu.get_act_cmd(),
							key_hash_, cc_, 2000-timeout_, step_, self._lock_id );
				end
				return self._lock_id   ---成功
			else
				if  err == 'exists' then
					---已经存在锁，等待
					timeout_ = timeout_ - step_
					if  timeout_ <= 0 then
						log.day_list_short("redis", "lock_rw", "error_time_out0",
								ngx.worker.id(), uu.get_act_cmd(), self.pre, redis_select,
								key_hash_, cc_, timeout_, step_ )
						return 0, "error_time_out_lock"    ---超时
					end

					if  cc_ <= 10 then
						step_ = 50 + ( cc_ * 2)   ---等待时间逐步放大   500+2+4+6+8+10+12+14+16+18+20=500 + 110
					else
						step_ = (20-cc_) * 10      ---等待时间逐步缩小  100 90 80 70 60 50 40 30
						if  step_ < 10 then step_ = 10 end
					end

					ngx_sleep( step_ * 0.001 )
				else
					log.day_list_short( "redis", "lock_rw", "error_step0",
							ngx.worker.id(), uu.get_act_cmd(), self.pre, redis_select,
							key_hash_, cc_, timeout_, step_, err );
					return  0, err  	---其他错误
				end
			end
		end    -- while wait

	else
		if  self.select_mod ~= 10 then
			---提示可能是百分表
			log.error( "error_select: plz check select_mod=" .. self.select_mod )
		end

		log.day_list_short("redis", "lock_rw", "error_no_redis_select", ngx.worker.id(), uu.get_act_cmd(),
				self.pre, redis_select, key_hash_ );
		return 0, "error_no_ngx_shared"   ---找不到 ngx_shared
	end

	return 0, "unknown_err"   ---未知错误 (应该不能运行到这里，健壮性保证)
end



---远程调用解锁，只有解锁后，其他【进程或协程】的请求才能继续获得读写锁，进行读写操作
---和check_redis_resty_lock是一对
---read_only_ 不解锁，等待final统一解锁
--- 0=成功   其他=失败
_M.unlock_resty_lock = function( self, redis_select, key_hash_, read_only_ )
	if  read_only_ then
		return 0    ---ok 不解锁
	end

	if  _REDIS_RESTY_LOCK then
		local dict_ = ngx_shared[ self.pre .. redis_select ]
		if  dict_ then
			local ok, err_ = dict_:delete( key_hash_ .. "_lk")
			if  ok then
				--删除成功
				--log.day_list_short( "redis", "lock_rw", "unlock_ok", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
				return 0     ---ok 成功
			else
				log.day_list_short( "redis", "lock_rw", "error_delete_unlock1", ngx.worker.id(),
						uu.get_act_cmd(), self.pre, redis_select, key_hash_, err_ );
				return 0     ----not_lock
			end
		else
			log.error( "dict_not_exist=" .. self.pre .. redis_select )
			log.day_list_short( "redis", "lock_rw", "error_dict_not_exist",  self.pre .. redis_select );
			return 1         ---"no_dict"
		end

	else
		return  0    ---ok --不校验
	end
end



------------读取数据
_M.API_redis_read = function( self, redis_select, key, hash, read_only_ )
	log.debug( log.fast() and "call API_redis_read, redis_select=" .. redis_select
			.. ", key=" .. key .. ", hash=" .. hash );
	self:set_get_set_tj( hash, "read_call" )

	local key_hash_ = key .. "_h" .. hash

	--乐观锁r
	if  read_only_ then
		--- 只读请求
		--log.day_list_short( "redis", "lock_rw", "read_only", ngx.worker.id(), uu.get_act_cmd(), key_hash_ );
	else
		if  _REDIS_RESTY_LOCK then
			local lock_id_, err_msg_ = self:check_redis_resty_lock(redis_select, key_hash_ )
			if  lock_id_ == 0 then
				self:error_exit( err_msg_ );
				return __FAIL__, err_msg_
			end
			ngx.ctx.__lock_id__ = lock_id_   ---记录lock_id
		end
	end

	local ret2, stale_ = self:get_shared_cache_redis( redis_select, key_hash_ )
	if  ret2 then
		if  stale_ then
			self:set_get_set_tj( hash, "read_cache_stale" )
		else
			self:set_get_set_tj( hash, "read_cache" )
		end
		return __SUCC__, ret2   --cache命中
	end

	--lock 模式
	local lock_id_ = self:read_lock2(redis_select)
	if  lock_id_  == 0 then
		self:error_exit( "read_lock_err" );
		return  __FAIL__, "read_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret, err = instance:hget(key, hash);
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hget_fail|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, err
	end

	if  type(ret) == 'string' then
		--- ok, 设置缓存 select_, key_, str_
		self:set_get_set_tj( hash, "read_ok" )
		if  ngx.ctx.no_shared_cache == 'y' then
			-- 不保存 scan使用
			log.day_list_short( "redis", "no_shared_cache", redis_select, key, hash );
		else
			if  #ret > 0 then
				self:set_shared_cache_redis( redis_select, key_hash_, ret )
			else
				log.day_list_simple( "redis", "error_len0", redis_select, key, hash, ret );
			end
		end

	elseif ret == ngx.null then
		ret = NGX_NULL
		self:set_get_set_tj( hash, "read_null" )
	else
		-- 返回的一定是一个string 或者 ngx.null 否则是出错
		log.day_list_short( "redis", "hget_fail", key, hash, "ret_not_string", type(ret) );
		self:error_exit( "hget err:" .. (err or "nil") );
		ngx.exit(ngx.ERROR);  	---严重出错退出
		return  __FAIL__, "ret_not_string"
	end

	return __SUCC__, ret;
end



---用来校验not_found的读取
_M.read_check_not_found = function( self, redis_select, key, hash )

	--lock 模式
	local lock_id_ = self:read_lock2(redis_select)
	if  lock_id_  == 0 then
		self:error_exit( "read_nil_lock_err" );
		return  __FAIL__, "read_nil_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret, err = instance:hget(key, hash);
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hget_fail|" .. key .. "|" .. hash  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, "hget_fail"
	end

	-- 处理返回值 返回是字符串或者NGX_NULL
	if  ret == ngx.null then
		ret = NGX_NULL
	elseif type(ret) == 'string' then
		--- ok
	else
		-- 返回的一定是一个table 否则是出错
		log.day_list_short( "redis", "hget_fail", key, hash, "ret_not_table", type(ret) );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, "ret_not_table"
	end

	return __SUCC__, ret
end



-- multi hget
_M.API_redis_mread = function( self, redis_select, all_kv_list )
	log.debug( log.fast() and "call API_redis_mread, redis_select=" .. redis_select );
	local fret = {};

	--先查cache
	local kv_list = {}           --需要redis的list
	for i=1, #all_kv_list do
		self.mread_tj.call = self.mread_tj.call + 1

		local key_hash_ = all_kv_list[i].key .. "_h" .. all_kv_list[i].hash
		local ret2, stale_ = self:get_shared_cache_redis( redis_select, key_hash_ )
		if  ret2 then
			--log.debug( log.fast() and "mread cache hit " .. key_hash_ )
			--log.day_list_short("redis", "cache_hit_redis", "mm|cache_hit|" .. key_hash_ )
			fret[i] = ret2;
			if  stale_ then
				self.mread_tj.gets = self.mread_tj.gets + 1
			else
				self.mread_tj.get =  self.mread_tj.get + 1
			end

		else
			--log.debug( log.fast() and "mread need redis " .. key_hash_ )
			--log.day_list_short("redis", "cache_hit_redis", "need_redis|" .. key_hash_ )
			--fret[i] = {}
			all_kv_list[i].__mi = i;     --mread index
			table_insert( kv_list, all_kv_list[i] )

			---只要触发流控，阻止请求redis，可以请求shared_cache
			if  self.read_only_redis_block > 0 then
				self.read_only_redis_block = self.read_only_redis_block - 1
				log.day_list_short( "redis", "mread|read_only_redis_block|" .. self.read_only_redis_block );
				self:error_exit( "mread err:read_only_redis_block" .. self.read_only_redis_block );
				return __FAIL__, "BLOCK"
			end

		end
	end


	local ret_list = {};

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "mread_lock_err" );
		return __FAIL__, "mread_lock_err"
	end

	local instance = self:try_connect(redis_select);
	instance:init_pipeline();
	for i=1, #kv_list do
		instance:hget( kv_list[i].key, kv_list[i].hash );
	end
	local ret_table_list_, err = instance:commit_pipeline();

	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "mread|" .. (err or "nil") );
		self:error_exit( "mread err:" .. (err or "nil") );
		return __FAIL__, err
	end

	--uu.var_dump( ret_table_list_ );

	if type(ret_table_list_) == 'table' then

		for i=1, #ret_table_list_ do
			local ret = ret_table_list_[i];

			local key_ = "nil";
			if  kv_list[i] then
				key_ = kv_list[i].key or "nil";
			end

			-- 处理返回值 返回是字符串或者NGX_NULL
			local type_ret_ = type(ret)
			if  type_ret_ == 'string' then
				---正常
			elseif  ret == ngx.null then
				ret = NGX_NULL;
			else
				---出错
				log.day_list_short( "redis", "mread", "ret_not_string", key_, type(ret) );
			end
			table_insert(ret_list, ret);
		end
	end


	--合并结果
	for i=1, #ret_list do
		if  kv_list[i] and kv_list[i].__mi then
			fret[ kv_list[i].__mi ] = ret_list[i]
		end
	end

	if  #fret ~= #all_kv_list then
		log.error("mread ret number not match")
		log.day_list_short( "redis", "cache_hit_redis", "error_count")
	end

	return __SUCC__, fret;

end



-- 删除一个key下的一个hash
_M.API_redis_del = function( self, redis_select, key, hash )
	log.debug( log.fast() and "call API_redis_del, redis_select=" .. redis_select
			.. ", key=" .. key .. ", hash=" .. hash );

	self:clean_shared_cache_redis( redis_select, key, hash )

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "del_lock_err" );
		return  __FAIL__, "del_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret_, err = instance:hdel(key, hash);
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hdel_fail", redis_select, key, hash, err)
		self:error_exit( "hdel err:" .. (err or "nil") );
		return __FAIL__, err
	end

	self:unlock_resty_lock( redis_select, key .. "_h" .. hash )  --key_hash_
	log.debug( log.fast() and "del:" .. uu.to_str(ret_) )

	return __SUCC__, ret_   ---被删除个数
end


-- 一个key下有多少个hash值
_M.API_redis_hlen = function( self, redis_select, key )
	log.debug( log.fast() and "call API_redis_hlen, redis_select=" .. redis_select .. ", key=" .. key );

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "hlen_lock_err" );
		return __FAIL__, "hlen_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret_, err = instance:hlen(key);
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hlen_fail|" .. redis_select .. "|" .. key  .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, err
	end

	return  __SUCC__, ret_
end



--遍历一个连接下的所有值
_M.API_redis_hkeys = function( self, redis_select, key_ )
	log.debug( log.fast() and "call API_redis_hkeys, start=" .. key_ );

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "hkeys_lock_err" );
		return __FAIL__, "hkeys_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret_table_, err = instance:hkeys( key_ );
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hkeys_fail|" .. redis_select .. "|" .. key_ .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, "hkeys_fail"
	end

	uu.var_dump( ret_table_ );
	return __SUCC__, ret_table_ or {};
end



--获得一个key下的所有值
_M.API_redis_hget_all = function( self, redis_select, key )
	log.debug( log.fast() and "call API_redis_hget_all, key=" .. key );

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "hget_all_lock_err" );
		return __FAIL__, "hget_all_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret_table_, err = instance:hgetall(key);
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "hlen_fail|" .. redis_select .. "|" .. key .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return __FAIL__, err
	end

	uu.var_dump( ret_table_ );
	return __SUCC__, ret_table_ or {};
end



--遍历一个连接下的所有值
_M.API_redis_scan = function( self, redis_select, cursor_ )
	log.debug( log.fast() and "call API_redis_scan, cursor=" .. cursor_ );

	--lock 模式
	local lock_id_ = self:read_lock2( redis_select )
	if  lock_id_  == 0 then
		self:error_exit( "scan_lock_err" );
		return __FAIL__, "scan_lock_err"
	end

	local instance = self:try_connect(redis_select);
	local ret_table_, err = instance:scan( cursor_ );
	self:read_unlock2(redis_select)

	if  err then
		log.day_list_short( "redis", "scan_fail|" .. redis_select .. "|" .. cursor_ .. "|" .. (err or "nil") );
		self:error_exit( "hget err:" .. (err or "nil") );
		return  __FAIL__, err
	end

	uu.var_dump( ret_table_ );
	return __SUCC__, ret_table_ or {};
end


--- API 清理掉一个cache数据
_M.API_redis_clean_shared_cache = function(self, p_select_, key_, hash_ )
	local select_ = p_select_ % self.select_mod
	self:clean_shared_cache_redis(select_, key_, hash_ )
end


--- API 尝试获得远程锁
_M.API_redis_try_get_resty_lock = function(self, p_select_, key_hash_ )
	local select_ = p_select_ % self.select_mod
	return self:check_redis_resty_lock(select_, key_hash_ )
end


--- API 尝试释放远程锁
_M.API_redis_unlock_resty_lock = function(self, p_select_, key_hash_ )
	local select_ = p_select_ % self.select_mod
	return self:unlock_resty_lock( select_, key_hash_ )
end



----------------------------------private functions-----------------------------------
--- 同进程信号锁semaphore lock
--- 相同worker下的不同协程同步 避免使用同一个socket句柄收发消息
--- limit_  流控系数，  ---优先保护写
------ return 0=未拿到锁(超限制 timeout)  1=拿到锁
_M.read_lock2 = function( self, pdb_name_, limit_ )
	local db_name_ = self.pre .. pdb_name_

	if  not self.read_semas_tj[db_name_] then
		self.read_semas_tj[db_name_] = {
			sema=ns_semaphore.new(0),   --锁
			task_cc = 0,             --任务数
		}
	end

	local db_sema_ = self.read_semas_tj[db_name_];
	local sema_count_ = db_sema_.sema:count()

	---检查任务数和加锁数差距是否过大
	local task_real_count_ = sema_count_ + db_sema_.task_cc
	if  task_real_count_ > 10 or task_real_count_ < -10 then
		log.day_list_short( "redis", "sema1", ngx.worker.id(),
				"task_real_count_warning", db_sema_.task_cc, sema_count_, db_name_ )
	end

	if  db_sema_.task_cc > (limit_ or __DEFAULT_READ_LIMIT__) then
		---超限制队列保护
		log.day_list_short( "redis", "sema2", ngx.worker.id(),
				"error_busy_w" .. db_sema_.task_cc, sema_count_, db_name_ )
		self.read_only_redis_block = __BUSY_BLOCK_LIMIT__   ---触发流控保护，无法mread
		return 0
	elseif  db_sema_.task_cc  < 1 then
		---第一个任务
		--if  sema_count_ ~= 0 then
		--log.day_list_short( "redis", "sema3", "w0_" .. db_sema_.task_cc, sema_count_, db_name_ )
		--end
		db_sema_.task_cc = db_sema_.task_cc + 1
		--ngx.sleep( 0.001 * math.random( 1, 100 ) )  		--haima test
		return 1     ---拿到锁
	else
		---需要排队的任务
		local start_   = ngx_now();

		db_sema_.task_cc = db_sema_.task_cc + 1
		local ok, err = db_sema_.sema:wait(5);
		if  ok then
			if  db_sema_.task_cc >= 10 then
				log.day_list_short( "redis", "sema4", ngx.worker.id(), db_name_,
						"w" .. db_sema_.task_cc, db_sema_.sema:count(), math.floor((ngx_now()-start_)*1000) )
			end

			--ngx.sleep( 0.001 * math.random( 1, 100 ) )  		--haima test
			return 1  ---正常唤醒 拿到锁
		else
			if  err then
				-- may timeout
				db_sema_.err_cc = (db_sema_.err_cc or 0) + 1
				log.day_list( "redis", "sema5", ngx.worker.id(), "error_" .. err, db_name_,
						"w" .. db_sema_.task_cc, db_sema_.sema:count(), math.floor((ngx_now()-start_)*1000) )
				log.error("ERROR: read_semaphore " .. err)
			end

			---任务强制结束 叫醒下一个sema
			db_sema_.task_cc = db_sema_.task_cc - 1
			if  db_sema_.sema:count() < 0 then
				db_sema_.sema:post(1)
			end

		end
	end

	return 0;
end



---信号锁解锁 semaphore unlock， 和read_lock2是一对
_M.read_unlock2 = function( self, pdb_name_ )

	local db_name_ = self.pre .. pdb_name_
	local db_sema_ = self.read_semas_tj[db_name_];
	if  db_sema_ then
		db_sema_.task_cc = db_sema_.task_cc - 1
		if  db_sema_.sema:count() < 0 then
			db_sema_.sema:post(1)
		end
	end
end



-- 出错处理
_M.error_exit = function(self, txt_)
	log.error( "redisMgrDc error_exit:" .. (txt_ or 'nil') );
	self:close(true);
end


-- 关闭所有REDIS连接
_M.close = function( self, shutdown )
	if  ngx.ctx.redis_dc_connections then
		for k, v in pairs( ngx.ctx.redis_dc_connections ) do
			self:private_close( v, k, shutdown );
		end
		ngx.ctx.redis_dc_connections = {};
	end
	--log.debug( log.fast() and "redis keep or close all ok." );
end


-- 关闭单个REDIS连接
_M.private_close = function( self, instance, redis_select, shutdown )
	local ok, err;
	if  shutdown == true then
		ok, err = instance:close();
		log.day_list_short( "redis",  "close|" .. redis_select );
	else
		ok, err = instance:set_keepalive( 120000, 256 );
	end

	if  err then
    	log.error("failed to set connection keepalive: " .. err .. ", redis_select=" .. redis_select );
    end
end


-- 按连接名字获得redis配置
_M.get_redis_config = function( self, redis_select_ )
	if  self.config_cache[ redis_select_ ]  then
		--先查config cache
		log.debug( log.fast() and "find config_cache for " .. redis_select_ );
		return  self.config_cache[ redis_select_ ];
	else
		--if  ns_config and ns_config.redis_db_select_s2 and ns_config.redis_server_s2 then
		--config_server = config_server_;
		--config_select = config_select_;
		if  self.config_server and self.config_select then
			if  self.config_select[redis_select_] then
				local ret_ = self.config_server[ self.config_select[redis_select_] ];
				if  ret_ then
					log.debug( log.fast() and "set config_cache for " .. redis_select_ );
					self.config_cache[ redis_select_ ] = ret_;
					return ret_;
				end
			end
		end
	end

	self:error_exit( 'error redis_select = ' .. (redis_select_ or 'nil') );
end



-- 按照select_name_, 获得REDIS连接
-- redis_select_ = comm, uin,
_M.try_connect = function( self, redis_select_ )
	self:init_pool();

	local config_ = self:get_redis_config( redis_select_ );

	ngx.ctx.redis_dc_connections_index =  (ngx.ctx.redis_dc_connections_index or 0) + 1
	local key_    = self.pre .. redis_select_ .. '_' .. ngx.ctx.redis_dc_connections_index

	if  ngx.ctx.redis_dc_connections[ key_ ] then
		log.debug( log.fast() and "use old redis connection. redis_select=" .. key_ );
		return ngx.ctx.redis_dc_connections[ key_ ];
	else
		--新建连接
		log.debug( log.fast() and "create new redis connection. redis_select=" .. key_ );
		local instance = self:private_try_connect( config_.ip, config_.port, config_.auth, redis_select_ );
		ngx.ctx.redis_dc_connections[ key_ ] = instance;
		return instance;
	end

end


-----------------------------------------------------------------------------utils---------------------
-- config = { ip, port, auth, select }
-- 建立redis连接
_M.private_try_connect = function( self, ip, port, auth, redis_select_ )

	local instance = ns_redis:new();
	instance:set_timeout(3000);      --3s

	--if  instance.set_timeouts then
	--	instance:set_timeouts(3000, 3000, 3000);   ---使用后会报错
	--else
	--	instance:set_timeout(3000);      --3s
	--end

	local ok, err = instance:connect( ip, port );
	if  err then
		self:read_unlock2(redis_select_)   		--- 唤醒下一个sema
		self:error_exit( "fail connect. msg=" .. err );
		return;
	end

	--log.day_list( "redis_socket", ip, port, uu.to_str( instance ), tostring(instance.sock),
			--ngx.worker.id(), instance:get_reused_times() )

	if  auth and #auth>0 then
		local count, err = instance:get_reused_times();
		if  0 == count then
			--set auth
			ok, err = instance:auth( auth );
			if  not ok then
				self:read_unlock2(redis_select_)   		--- 唤醒下一个sema
				self:error_exit( "fail auth. msg=" .. err );
				return;
			end

			ngx.ctx.redis_reused_count = 0;
		elseif err then
			self:read_unlock2(redis_select_)   		--- 唤醒下一个sema
			self:error_exit( "fail get_reused_times. msg=" .. err );
			return;
		else
			log.debug( log.fast() and "get_reused_times redis=" .. count );
			ngx.ctx.redis_reused_count = count;
		end

		--ok, err = instance:select(0);  ---默认都使用第0个
		--if  err then
			--self:error_exit( "fail select. msg=" .. err );
		--end
	end

	return instance;
end



--写校验[_k_] [_h_]
_M.private_check_set_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			log.day_list_short( "redis", "error_key_check", "set", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_set_key_check" )
			return false;
		end
	else
		value._k_ = tonumber(key) or key;  --首次设置
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			log.day_list_short( "redis", "error_hash_check", "set", db_name_, key, value._h_, hash );
			log.error( "error_set_hash_check" )
			return false;
		end
	else
		if  tonumber(hash) then
			value._h_ = tonumber(hash) --首次设置
		end
	end

	value._t_ = uu.now()    --最后一次写入时间
	return true;
end



--读校验[_k_] [_h_]
_M.private_check_get_key = function(self, db_name_, key, hash, value )
	if  value._k_ then
		if  ( tonumber(value._k_) and ( tonumber(key) == tonumber(value._k_) ) ) or (key == value._k_) then
			--ok
		else
			--读取出错
			log.day_list_short( "redis", "error_key_check", "get", db_name_, key, value._k_, hash, debug.traceback() );
			log.error( "error_get_key_check" )
			return false;
		end
	end

	if  value._h_ then
		if  ( tonumber(value._h_) and ( tonumber(hash) == tonumber(value._h_) ) ) or (hash == value._h_) then
			--ok
		else
			--读取出错
			log.day_list_short( "redis", "error_hash_check", "get", db_name_, key, value._h_, hash );
			log.error( "error_get_hash_check" )
			return false;
		end
	end

	return true;
end




-----------------------------ngx_shared--------------------------------

--设置全局数据缓存kv 十分表
_M.set_shared_cache_redis = function( self, select_, key_, str_, expire_ )
	if  select_ and key_ and str_  then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			if  expire_ and type(expire_) == 'string' then
				--log.debug( log.fast() and "expire_=" .. expire_ .. " k=" .. key_ .. " v=" .. str_ )
				expire_ = tonumber(expire_) or 0
			end

			if  expire_ and expire_ >=0 and expire_ < 7200 then
				--normal
			else
				expire_ = 7200     --2*60*60
			end

			local ok, err, forcible = share_:set( key_, str_, expire_ );
			if  ok then
				if  forcible then
				    log.day_list_short("redis", "cache_hit_redis", "set_forcible", dict_name_, key_, err )
				end
				return true
			else
				if  err then
					--share_:delete( key_ )
					--set出错，立即删除缓存
					local ok, err, forcible = share_:set( key_, nil );
					if  err or forcible then
						log.day_list_short("redis", "cache_hit_redis", "error_delete", dict_name_, key_, err, forcible )
					end
				end
				log.day_list_short("redis", "cache_hit_redis", "error_set", dict_name_, key_, err )
			end
		else
			log.day_list_short("redis", "cache_hit_redis", "error_set_no_dict", dict_name_, key_)
		end
	end

	return false   --写入cache失败
end



--获得全局数据缓存kv 十分表
_M.get_shared_cache_redis = function( self, select_, key_ )
	if  select_ and key_ then

		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then

			local str_, f_, stale_ = share_:get_stale( key_);
			--uu.report_share_free_space( share_, dict_name_, key_, "redis" )

			if  str_ then
				return str_, stale_  	--正常返回
			else
				--有报错的情况 ret = nil, error
				if  f_ then
					log.day_list_short("redis", "cache_hit_redis", "error_get_nil", dict_name_, key_, f_ )
				end
			end
		else
			log.day_list_short("redis", "cache_hit_redis", "error_get_no_dict", dict_name_, key_)
		end
	end

	return nil, nil
end




--清理掉一个cache数据 十分表
_M.clean_shared_cache_redis = function(self, select_, key_, hash_ )
	if  select_ and key_ and hash_ then
		local dict_name_
		if  type(select_) == 'number' then
			dict_name_ = self.pre .. (select_ % self.select_mod)
		else
			dict_name_ = self.pre ..  select_
		end

		local share_ = ngx_shared[ dict_name_ ];
		if  share_ then
			local ok, err, forcible = share_:set( key_ .. "_h" .. hash_ , nil );
			if  err or forcible then
				log.day_list_short("redis", "cache_hit_redis", "error_delete", dict_name_, key_, err, forcible )
			end
		end
	end
end



return _M;
